/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.ram;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.impl.ReadOnlyChannel;
import org.sabayframework.channel.impl.WriteOnlyChannel;

/**
 * FanOutChannel is a concurrent, write-only channel that propagates 
 * each element to multiple read-only channels.
 * No limit on size of the channel or lock, limited only by available RAM.
 * @author Alex Antonau
 * */
public class FanOutChannel<T> extends WriteOnlyChannel<T> {

	private static class Node{
		Object item;
		Node next;
	}
	private Node first;
	private Node last;
    /** Number of items in the queue */
    private volatile int count;
    private final int capacity;
    /** closed flag */
    private volatile boolean closed;
    private final Condition notEmpty;

    /** Main lock guarding all access */
    private final ReentrantLock lock;
    
    private ArrayList<OutputChannel> outputs;
    
    public FanOutChannel(int numberOfReadChannels){
    	this(Integer.MAX_VALUE, numberOfReadChannels);
    }
    
	public FanOutChannel(int capacity, int numberOfReadChannels) {
		this.capacity = capacity;
        this.lock = new ReentrantLock();
        this.outputs = new ArrayList<OutputChannel>(numberOfReadChannels);
        for(int i=0;i<numberOfReadChannels;++i)outputs.add(new OutputChannel());
        notEmpty = lock.newCondition();
	}
	
	private Node firstForSize(int size){
		int index = count - size;
		Node n = first;
		while(index-->0) n = n.next;
		return n;
		
	}
	private void cleanNodes(){
		int maxCount = 0;
		for(int i=0;i<outputs.size();++i)
			if(outputs.get(i).count > maxCount) maxCount = outputs.get(i).count;
		int toClean = count - maxCount;
		count = maxCount;
		if(count == 0) first = last = null;
		else{
			Node n = first;
			while(toClean-->0) n = n.next;
			first = n;
		}
		
	}
	final class OutputChannel extends ReadOnlyChannel<T>{
		private volatile int count;
		private final class ReadLock implements Lock<T>{
	    	private int locked;
	    	private Node first;
			ReadLock(final int size) {
				first = firstForSize(count);
				count -= size;
				locked = size;
			}
			@Override
			public int size() {
				return locked;
			}

			@SuppressWarnings("unchecked")
			@Override
			public T get(int index) {
				if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
				Node n = first;
				while(index-->0) n = n.next;
				return (T) n.item;
			}
			@Override
			public void set(int index,final T elem) {
				raiseReadOnlyException();
			}
			@Override
			public Object[] toArray() {
				Object[] res = new Object[locked];
				Node n = first;
				for(int i=0;i<locked;++i){
					res[i] = n.item;
					n = n.next;
				}
				return res;
			}
			@Override
			public void unlock() {
				if(locked!=-1){
					cleanNodes();
					locked = -1;
					lock.unlock();
				}
			}
			@Override
			public void cancel() {
				if(locked != -1){
					count+=locked;
					locked = -1;
					lock.unlock();
				}
			}
			@Override
			protected void finalize() throws Throwable {
				unlock();
			}
	    }
		
		@Override
		public Lock<T> getReadLock(final int size) {
			if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
			ReadLock result = null;
			if(lock.tryLock()){
				try{
					if(size <= count){
						result = new ReadLock(size);
					}
				} finally {
					if(result == null) lock.unlock();
				}
			}
			return result;
		}

		@Override
		public boolean isEmpty() {
			return count == 0;
		}
		
		@Override
		public int size() {
			return count;
		}

		@Override
		public void close() {
			FanOutChannel.this.close();
		}

		@Override
		public boolean isClosed() {
			return FanOutChannel.this.isClosed();
		}

		@Override
		public void fail(Throwable cause) {
			FanOutChannel.this.fail(cause);
		}

		@Override
		public boolean hasFailed() {
			return FanOutChannel.this.hasFailed();
		}

		@Override
		public Throwable getFailureCause() {
			return FanOutChannel.this.getFailureCause();
		}

		@Override
		public Lock<T> waitForReadLock(int size, long timeout)
				throws InterruptedException {
	        final ReentrantLock lock = FanOutChannel.this.lock;
	        long time=System.currentTimeMillis();
	        ReadLock readLock = null;
	        if(lock.tryLock(timeout, TimeUnit.MILLISECONDS)){
		        try {
	            	timeout -= (System.currentTimeMillis()-time);
		            try {
		                while (timeout > 0 && !closed && count < size){
		                	time=System.currentTimeMillis();
		                	notEmpty.await(timeout, TimeUnit.MILLISECONDS);
		                	timeout -= (System.currentTimeMillis()-time);
		                }
		            } catch (InterruptedException ie) {
		            	notEmpty.signal(); // propagate to non-interrupted thread
		                throw ie;
		            }
					if(size <= count && !closed){
						readLock = new ReadLock(size);
						return readLock;
					}
		            
		        } finally {
		            if(readLock == null) lock.unlock();
		        }
	        }
	        return getReadLock(size);
		}
		
	}
	
	/**
	 * Returns a read-only channel by index.
	 * */
	public Channel<T> getReadChannel(int index){
		return outputs.get(index);
	}
    private void addItem(T item){
    	if(first == null){
    		first = new Node();
    		first.item = item;
    		last = first;
    	}else{
			last.next = new Node();
			last = last.next;
			last.item = item;
    	}
    }
	private final class WriteLock implements Lock<T>{
		private int locked;
		private Node first;
		
		WriteLock(int size) {
			locked = size;
			Node oldLast = last;
			while(size-->0)addItem(null);
			first = (oldLast == null)? 
					FanOutChannel.this.first:
					oldLast.next;
			count+=locked;
		}
		@Override
		public int size() {
			return locked;
		}
		@SuppressWarnings("unchecked")
		@Override
		public T get(int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			return (T) n.item;
		}
		@Override
		public void set(int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			n.item = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			Node n = first;
			for(int i=0;i<locked;++i){
				res[i] = n.item;
				n = n.next;
			}
			return res;
		}

		@Override
		public void unlock() {
			if(locked!=-1){
				for(OutputChannel ouch: outputs) ouch.count += locked;
				locked = -1;
				first = null;
				notEmpty.signalAll();
				lock.unlock();
			}
		}
		@Override
		public void cancel() {
			if(locked!=-1){
				count-=locked;
				if(count ==0){
					FanOutChannel.this.first = null;
					FanOutChannel.this.last = null;
				}else if(locked > 0){
					Node n = FanOutChannel.this.first;
					while(n.next != first) n=n.next;
					if(n!=null) {
						n.next = null;
						FanOutChannel.this.last = n;
					}
				}
				locked = -1;
				lock.unlock();
			}
		}
	}
	@Override
	public Lock<T> getWriteLock(final int size) {
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		WriteLock result = null;
		if(lock.tryLock()){
			try{
				if(size <= freeSize()){
					result = new WriteLock(size);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}
	
	@Override
	public boolean isFull() {
		return count >= capacity();
	}

	@Override
	public int size() {
		return count;
	}

	@Override
	public int capacity() {
		return capacity;
	}

	@Override
	public int freeSize() {
		if(closed)return 0;
		return capacity()-count;
	}

	@Override
	public boolean isEmpty() {
		return count == 0;
	}

	@Override
	public void close() {
		if(closed) return;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
       		closed = true;
       		notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

}
